package com.arelikebrothers.demo.kafka.stream;

import com.arelikebrothers.demo.kafka.stream.bean.InputMessage;
import com.arelikebrothers.demo.kafka.stream.bean.OutputMessage;
import com.google.gson.Gson;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.KeyValueStore;

/**
 * Created by lennon on 17/07/2017.
 */
public class DemoProcessor implements Processor<String, String> {
    private ProcessorContext context;
    private KeyValueStore<String, String> store;

    @Override
    public void init(ProcessorContext context) {
        this.context = context;
        this.context.schedule(1000);

        store = (KeyValueStore<String, String>) this.context.getStateStore("RESULT");
    }

    Gson gson = new Gson();

    @Override
    public void process(String key, String value) {
        System.out.println(key + "==>" + value);
        if (value == null) {
            context.commit();
            return;
        }

        InputMessage inputMessage = gson.fromJson(value, InputMessage.class);
        String id = inputMessage.getKey1() + "-" + inputMessage.getKey2();
        OutputMessage outputMessage = new OutputMessage(id, inputMessage.getContent());

        store.put(id, gson.toJson(outputMessage));
        context.commit();

    }

    @Override
    public void punctuate(long l) {
        System.out.println("====");
        KeyValueIterator<String, String> all = store.all();
        while (all.hasNext()) {
            KeyValue<String, String> next = all.next();
            System.out.println(next.key + "--->" + next.value);
            context.forward(next.key, next.value);
        }
        all.close();
        context.commit();
    }

    @Override
    public void close() {
        store.close();
    }
}
